package com.didispace.service;

import static org.junit.Assert.*;

import java.util.ArrayList;
import java.util.List;import java.util.concurrent.TimeUnit;

import org.junit.Test;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.observables.GroupedObservable;

public class TestRxjavaObserve1 {

	@Test
	public void test1() {
		Observable<String> myObservable = Observable.create(  
			    new Observable.OnSubscribe<String>() {  
			        @Override  
			        public void call(Subscriber<? super String> sub) {  
			            sub.onNext("Hello, eshin!");  
				            sub.onCompleted();  
				        }
				    }  
				);  
		Subscriber<String> mySubscriber = new Subscriber<String>() {  
			@Override  
			public void onNext(String s) { System.out.println(s); }  
			
			@Override  
			public void onCompleted() { }  
			
			@Override  
			public void onError(Throwable e) { }  
		}; 
		
		myObservable.subscribe(mySubscriber);  
	}
	
	@Test
	public void test2() {
		Observable.just("hello eshin")
		.map(new Func1<String, String>() {

			@Override
			public String call(String t) {
				return "call:" + t;
			}
		})
		.map(new Func1<String, Integer>() {

			@Override
			public Integer call(String t) {
				System.out.println(t);
				return t.length();
			}
		})
		.subscribe(new Action1<Integer>() {

			@Override
			public void call(Integer t) {
				System.out.println(t);
			}
		});
	}
	
	@Test
	public void test3() {
		Observable.from(new String[]{"hello","hi"}).subscribe(new Action1<String>() {

			@Override
			public void call(String t) {
				System.out.println(t);
			}
		});
		
		Observable<Integer> obs = Observable.just("hello eshin").flatMap(new Func1<String, Observable<Integer>>() {

			@Override
			public Observable<Integer> call(String t) {
				return Observable.just(t.length());
			}
		});

		obs.subscribe(new Subscriber<Integer>() {

			@Override
			public void onCompleted() {
				System.out.println("finish");
			}

			@Override
			public void onError(Throwable e) {
				System.out.println(e.getMessage());
			}

			@Override
			public void onNext(Integer t) {
				System.out.println("length:"+t);
			}

		});
	
		
		obs.concatMap(new Func1<Integer, Observable<String>>() {

			@Override
			public Observable<String> call(Integer t) {
				String length = "length is :" +t;
				return Observable.just(length);
			}
		}).subscribe(new Action1<String>() {

			@Override
			public void call(String t) {
				System.out.println(t);
			}
		});
	
	}
	@Test
	public void test4() {
		Observable<String> getObservable = getExecutionObservable();
		getObservable.subscribe(new Action1<String>() {

			@Override
			public void call(String t) {
				System.out.println(t);
				
			}
		});
		getObservable.subscribe();
	}
	@Test
	public void test5() {
		/**
		 * Scan操作符对原始Observable发射的第一项数据应用一个函数，
		 * 然后将那个函数的结果作为自己的第一项数据发射。它将函数的
		 * 结果同第二项数据一起填充给这个函数来产生它自己的第二项数据。
		 * 它持续进行这个过程来产生剩余的数据序列。这个操作符在某些情
		 * 况下被叫做accumulator
		 */
		Integer[]items = {1,2,3,4,5,6};  
		  
	       Observable<Integer> myObservable = Observable.from(items)  
	               .scan(new Func2<Integer, Integer, Integer>() {  
	                   @Override  
	                   public Integer call(Integer integer, Integer integer2) {  
	                       return integer + integer2;  //
	                   }  
	               });  
	  
	  
	       Subscriber<Integer> mySubscriber = new Subscriber<Integer>() {  
	           @Override  
	           public void onNext(Integer s) {  
	               System.out.println("onNext................."+s);  
	           }  
	  
	           @Override  
	           public void onCompleted() {  
	               System.out.println("onCompleted.................");  
	           }  
	  
	           @Override  
	           public void onError(Throwable e) {  
	               System.out.println("onError....................");  
	           }  
	       };  
	  
	       myObservable.subscribe(mySubscriber);  
	}
	@Test
	public void test6() throws InterruptedException {
		List<Integer>items = new ArrayList<>();  
		items.add(2);
		items.add(3);
		items.add(5);
		Thread.sleep(1000L);
		items.add(7);
		items.add(9);
		Thread.sleep(1000L);
		items.add(10);
		items.add(11);
	    Observable<Observable<Integer>> myObservable = Observable.from(items)  
							    		                                  .window(300, TimeUnit.MICROSECONDS)
							    		                                  .share()
							    		                                  .onBackpressureDrop() ;
	      Action1<Observable<Integer>>  action1 = new Action1<Observable<Integer>>(){  
	  
	          @Override  
	          public void call(Observable<Integer> integerObservable) {  
	              integerObservable.subscribe(new Action1<Integer>() {  
	                  @Override  
	                  public void call(Integer t) {  
								System.out.println("接收到的值是............."+t);  
	                  }  
	              });  
	              System.out.println("=======================================");
	          }  
	      }; 
	  
//	      new Thread(new Runnable() {
//			
//			@Override
//			public void run() {
//				while(true){
//					myObservable.subscribe(action1); 
//					try {
//						Thread.sleep(4000);
//					} catch (InterruptedException e) {
//						// TODO Auto-generated catch block
//						e.printStackTrace();
//					}
//				}
//			}
//		}).start();
	      
	      new Thread(new Runnable() {
	    	  
	    	  @Override
	    	  public void run() {
	    		  while(true){
	    			  items.add((int) System.currentTimeMillis());
	    			  try {
	    				  Thread.sleep(1000);
	    			  } catch (InterruptedException e) {
	    				  // TODO Auto-generated catch block
	    				  e.printStackTrace();
	    			  }
	    		  }
	    	  }
	      }).start();
	      myObservable.subscribe(action1);
	      while (true) {
//	    	Thread.sleep(2000L);
		} 
	}
	@Test
	public void test61() throws InterruptedException {
		ObservableBean observableBean  = new ObservableBean();
		observableBean.setName("hello");
		Observable<Observable<ObservableBean>> myObservable = Observable.create(new Observable.OnSubscribe<ObservableBean>() {
            @Override
            public void call(Subscriber<? super ObservableBean> subscriber) {
                while (true) {
                	observableBean.setTimeStamp(""+System.currentTimeMillis());
                    subscriber.onNext(observableBean);
                    try {
						Thread.sleep(2000L);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}//每隔2s发送消息
                }

            }
        })  
				.window(6,3)
				/*.share()
				.onBackpressureDrop()*/ ;
		Action1<Observable<ObservableBean>>  action1 = new Action1<Observable<ObservableBean>>(){  
			
			@Override  
			public void call(Observable<ObservableBean> integerObservable) {  
				System.out.println("=======================================");
				integerObservable.subscribe(new Action1<ObservableBean>() {  
					@Override  
					public void call(ObservableBean t) {  
						System.out.println("接收到的值是............."+t.toString());  
					}  
				});  
//				System.out.println("=======================================");
			}  
		}; 
		
		myObservable.subscribe(action1);
		while (true) {
//	    	Thread.sleep(2000L);
		} 
	}
	@Test
	public void test63() throws InterruptedException {
		ObservableBean observableBean  = new ObservableBean();
		observableBean.setName("hello");
		Observable<Observable<ObservableBean>> myObservable = Observable.create(new Observable.OnSubscribe<ObservableBean>() {
			@Override
			public void call(Subscriber<? super ObservableBean> subscriber) {
				while (true) {
					observableBean.setTimeStamp(""+System.currentTimeMillis());
					subscriber.onNext(observableBean);
					try {
						Thread.sleep(2000L);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}//每隔2s发送消息
				}
				
			}
		})  
				.window(8,TimeUnit.SECONDS)
				/*.share()
				.onBackpressureDrop()*/ ;
		Action1<Observable<ObservableBean>>  action1 = new Action1<Observable<ObservableBean>>(){  
			
			@Override  
			public void call(Observable<ObservableBean> integerObservable) {  
				System.out.println("=======================================");
				integerObservable.subscribe(new Action1<ObservableBean>() {  
					@Override  
					public void call(ObservableBean t) {  
						System.out.println("接收到的值是............."+t.toString());  
					}  
				});  
//				System.out.println("=======================================");
			}  
		}; 
		
		myObservable.subscribe(action1);
		while (true) {
//	    	Thread.sleep(2000L);
		} 
	}
	@Test
	public void test62() throws InterruptedException {
		Integer[]items = {1,2,3,4,5,6,7,8,9,10,11,12,13};  
		  
	      Observable<Observable<Integer>> myObservable = Observable.from(items).window(2,4);  
//	      Observable<Observable<Integer>> myObservable = Observable.from(items).window(3);  
	  
	      Action1<Observable<Integer>>  action1 = new Action1<Observable<Integer>>(){  
	  
	          @Override  
	          public void call(Observable<Integer> integerObservable) {  
	              integerObservable.subscribe(new Action1<Integer>() {  
	                  @Override  
	                  public void call(Integer integer) {  
	                      System.out.println("接收到的值是............."+integer);  
	                  }  
	              });  
	              System.out.println("=======================================");
	          }  
	      };  
	  
	      myObservable.subscribe(action1); 
		
	}
	
	/**
	 * GroupBy操作符将原始的Observerble按照某种规则拆分成一些Observables集合，
	 * 返回Observable的子类GroupedObservable,这个对象有一个方法getKey(),
	 * 可以根据这个判断在哪个分组中。
	 */
	@Test
	public void test7() {
		String[]items = {"GroupBy1","IGroupBy1","IGroupBy1","GroupBy1","GroupBy1","IGroupBy1"};  
		  
	       Observable<GroupedObservable<Boolean,String>> myObservable = Observable.from(items)  
	               .groupBy(new Func1<String, Boolean>() {  
	                   @Override  
	                   public Boolean call(String s) {  
	                     return s.startsWith("I");  
	                   }  
	               });  
	  
	       Action1  action1 = new Action1<GroupedObservable<Boolean,String>>(){  
	  
	           @Override  
	           public void call(final GroupedObservable<Boolean, String> booleanStringGroupedObservable) {  
	               booleanStringGroupedObservable.subscribe(new Action1<String>() {  
	                   @Override  
	                   public void call(String s) {  
	                       System.out.println("key..........startsWith I:" + booleanStringGroupedObservable.getKey() + "............" + s);  
	                   }  
	               });  
	  
	           }  
	       };  
	  
	       myObservable.subscribe(action1);  
	}
	
	/**
	 * buffer(count)将原来的observable按照参数count组成长度为count的集合，以集合的形式发射出去
	 */
	@Test
	public void test8() {
		Integer[]items = {1,2,3,4,5,6,7,8};  
		  
	      Observable<List<Integer>> myObservable = Observable.from(items)  
	              .buffer(3);  
	  
	  
	      Subscriber<List<Integer>> mySubscriber =new Subscriber<List<Integer>>() {  
	          @Override  
	          public void onNext(List<Integer> list) {  
	              System.out.println("onNext集合的长度是................."+list.size());  
	              for(Integer i:list){  
	                  System.out.println("onNext循环................."+i);  
	              }  
	          }  
	  
	          @Override  
	          public void onCompleted() {  
	              System.out.println("onCompleted.................");  
	          }  
	  
	          @Override  
	          public void onError(Throwable e) {  
	              System.out.println("onError....................");  
	          }  
	      };  
	  
	      myObservable.subscribe(mySubscriber);  

	}
	@Test
	public void test9() {
		
	}
	@Test
	public void test10() {
		Observable.interval(100,TimeUnit.MILLISECONDS).take(100).window(3,TimeUnit.SECONDS).subscribe(new Observer<Observable<Long>>() {  
            @Override  
            public void onCompleted() {  
            }  
  
            @Override  
            public void onError(Throwable e) {  
            }  
  
            @Override  
            public void onNext(Observable<Long> integerObservable) {  
            	System.out.println("==========================================");
                integerObservable.subscribe(new Action1<Long>() {  
                    @Override  
                    public void call(Long integer) {  
                        System.out.println(integer);
                    }  
                });  
                
            }  
        });
		
		while (true) {
//	    	Thread.sleep(2000L);
		}
	}
	@Test
	public void test11() {
		    
		Observable<Observable<String>> ob  =  Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                while (true) {
                    subscriber.onNext("消息" + System.currentTimeMillis());
                    try {
						Thread.sleep(2000L);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}//每隔2s发送消息
                }

            }
        }).window(6,TimeUnit.SECONDS).share().onBackpressureDrop();
		ob.window(3, 1)
//		.flatMap(new Func1<Observable<Observable<Long>>, Observable<Long>>() {
//
//			@Override
//			public Observable<Long> call(Observable<Observable<Long> t) {
//				// TODO Auto-generated method stub
//				return null;
//			}
//		});
		          
		    .subscribe(new Subscriber<Observable<Observable<String>>>() {  
			@Override  
			public void onCompleted() {  
			}  
			
			@Override  
			public void onError(Throwable e) {  
			}  
			
			@Override  
			public void onNext(Observable<Observable<String>> integerObservable) {  
				System.out.println("==========================================");
				
				integerObservable.subscribe(new Action1<Observable<String>>() {  
					@Override  
					public void call(Observable<String> integer) {  
						System.out.println("-------------------------------------------");
						integer.subscribe(new Action1<String>() {

							@Override
							public void call(String t) {
								System.out.println(t);
							}
						});
					}  
				});  
				
			}  
		});
		
		while (true) {
//	    	Thread.sleep(2000L);
		}
	}

	public static Observable<String> getExecutionObservable() {
        return Observable.defer(new Func0<Observable<String>>() {
            @Override
            public Observable<String> call() {
                try {
                    return Observable.just("hello game");
                } catch (Throwable ex) {
                    return Observable.error(ex);
                }
            }
        }).doOnSubscribe(new Action0() {
            @Override
            public void call() {
                // Save thread on which we get subscribed so that we can interrupt it later if needed
               System.out.println("game over0");
            }
        }).doOnSubscribe(new Action0() {
        	@Override
        	public void call() {
        		// Save thread on which we get subscribed so that we can interrupt it later if needed
        		System.out.println("game over1");
        	}
        }).doOnSubscribe(new Action0() {
        	@Override
        	public void call() {
        		// Save thread on which we get subscribed so that we can interrupt it later if needed
        		System.out.println("game over2");
        	}
        }).doOnSubscribe(new Action0() {
			
			@Override
			public void call() {
				System.out.println("game over3");
			}
		});
    }
}
